package com.sisyphus.source

import java.util.Properties

import com.sisyphus.utils.ParamsConf
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

/**
 * kafka数据源
 */
class KafkaSource {
  def flinkKafkaConsumer(): FlinkKafkaConsumer[String] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", ParamsConf.kafkaHosts)
    properties.setProperty("group.id", ParamsConf.kafkaGroupId)

    new FlinkKafkaConsumer[String](ParamsConf.kafkaTopic, new SimpleStringSchema(), properties)
  }
}
